package ysgrpc

import (
	"context"
	"gitee.com/kmyss/gf-ex/ysgrpc/proto"
	"gitee.com/kmyss/gf-ex/yslog"
	"github.com/gogf/gf/v2/container/garray"
	"github.com/gogf/gf/v2/container/gmap"
	"github.com/gogf/gf/v2/encoding/gjson"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/text/gregex"
	"github.com/gogf/gf/v2/text/gstr"
	"github.com/gogf/gf/v2/util/gconv"
	"io"
	"time"
)

const cfgAll = "__all"

type MQModuleCfg struct {
	Name             string `json:"name"`            // 名称
	UseHeartHandler  bool   `json:"useHeartHandler"` // 心跳
	HasSubAllTopic   bool   `json:"hasSubAllTopic"`  //
	UseLoggerHandler bool   `json:"useLoggerHandler"`
	UseUnPack        bool   `json:"useUnPack"`
	UseConfigCenter  bool   `json:"useConfigCenter"`
}

// NewDefaultMQModuleCfg 生成默认的 MQModule 配置参数
func NewDefaultMQModuleCfg() MQModuleCfg {
	return MQModuleCfg{
		UseHeartHandler: true,
		UseUnPack:       true,
		UseConfigCenter: true,
	}
}

func MQEmptyJsonMessage() []byte {
	return gconv.Bytes("{}")
}

type MQHandlerFunc = func(topic string, msg []byte)

type IMQModule interface {
	Cfg() MQModuleCfg
	Init(ctx context.Context)
	Run()
	Stop()
}

type MQBaseModule struct {
	l   *glog.Logger
	cfg MQModuleCfg
	cl  *MQClient
	ctx context.Context

	topics   *garray.StrArray // 存储话题数据
	handlers *gmap.ListMap
	id       int32
	packs    *gmap.StrAnyMap // 存储包文件
	hasRun   bool

	TopicTool *MQTopicTool
	AppCfg    *gjson.Json
}

func (m *MQBaseModule) Cfg() MQModuleCfg {
	return m.cfg
}

func (m *MQBaseModule) SetCfg(cfg MQModuleCfg) {
	m.cfg = cfg
}

func (m *MQBaseModule) Log() *glog.Logger {
	if m.l == nil {
		m.l = yslog.Log(m.cfg.Name)
	}
	return m.l
}

func (m *MQBaseModule) Client() *MQClient {
	return m.cl
}

func (m *MQBaseModule) Init(ctx context.Context) {
	// 内部参数初始化
	m.TopicTool = NewMQTopicTool(m.cfg.Name)
	m.cl = NewMQClient(ctx)
	m.ctx = ctx

	m.topics = garray.NewStrArray()
	m.handlers = gmap.NewListMap()
	m.id = -1
	m.packs = gmap.NewStrAnyMap()

	if m.cfg.UseHeartHandler {
		m.AddTopic(m.TopicTool.HEART_ASK)
		m.AddHandlerFunc(m.TopicTool.GenHeartTopic(MQTopicInput{}), m.heartHandler)
	}

	if m.cfg.UseLoggerHandler {
		m.Log().SetHandlers(m.logHandler)
	}

	if m.cfg.UseConfigCenter {
		m.AddTopic(m.TopicTool.GenCfgTopic(MQTopicInput{}))
		m.AddTopic(m.TopicTool.GenCfgTopic(MQTopicInput{Name: cfgAll}))
		m.AddHandlerFunc(m.TopicTool.CFG, m.cfgHandler)
	}

	m.AddHandlerFunc("ok.sub", m.okSubHandler)
}

func (m *MQBaseModule) AddHandlerFunc(topic string, handler MQHandlerFunc) {

	// 判断话题是否存在！
	if _, ok := m.handlers.Search(topic); ok {
		m.handlers.Remove(topic)
		m.Log().Warningf(m.ctx, "%v 已存在！处理函数被覆盖", topic)
	}

	m.handlers.Set(topic, handler)

}

func (m *MQBaseModule) DeleteHandlerFunc(topic string, handler MQHandlerFunc) {

	// 判断话题是否存在！
	if _, ok := m.handlers.Search(topic); ok {
		m.handlers.Remove(topic)
	} else {
		m.Log().Warningf(m.ctx, "%v 不存在！", topic)
	}

}

func (m *MQBaseModule) AddTopic(topic string) {

	// 判断话题是否存在！
	if index := m.topics.Search(topic); index == -1 {
		m.topics.Append(topic)
		_ = m.UnSub()
	} else {
		m.Log().Warningf(m.ctx, "%v 已存在！", topic)
	}

}

func (m *MQBaseModule) DeleteTopic(topic string) {
	if index := m.topics.Search(topic); index != -1 {
		m.topics.Remove(index)
		_ = m.UnSub()
	} else {
		m.Log().Warningf(m.ctx, "%v 不存在！", topic)
	}
}

// Run 阻塞运行模块
func (m *MQBaseModule) Run() {
	var (
		sub proto.MQ_SubClient
		err error
	)

	m.hasRun = true
	for m.hasRun {
		sub, err = m.cl.Sub(m.genSubTopics())
		if err != nil {
			m.Log().Warning(m.ctx, "连接服务器超时，2后重试！")
			time.Sleep(2 * time.Second)
			continue
		} else {
			m.Log().Info(m.ctx, "订阅成功")
		}

		m.askCfg()

		for err == nil {
			var recv *proto.Message
			recv, err = sub.Recv()
			if err == io.EOF {
				m.Log().Info(m.ctx, "订阅取消！")
				break
			} else if err != nil {
				if m.id < 0 {
					m.Log().Info(m.ctx, "订阅取消！")
				} else {
					m.Log().Error(m.ctx, "连接中断！")
				}
				break
			}

			// 解包分组数据
			topic := recv.GetTopic()
			message := recv.GetMessage()
			if m.cfg.UseUnPack {
				topic, message = m.doUnpack(topic, message)
				if g.IsEmpty(topic) {
					continue
				}
			}

			// 调用处理函数
			m.Log().Debug(m.ctx, "接收话题", topic)
			for _, key := range m.handlers.Keys() {
				k := key.(string)
				if gstr.HasPrefix(topic, k) {
					go m.handlers.Get(key).(MQHandlerFunc)(topic, message)
				}
			}

		}

	}
}

func (m *MQBaseModule) Stop() {
	if m.hasRun {
		m.hasRun = false
		_ = m.UnSub()
	}

	if m.cl != nil {
		_ = m.cl.Close()
	}

}

func (m *MQBaseModule) UnSub() error {
	if m.id != -1 {
		err := m.cl.UnSub(m.id)
		if err != nil {
			return err
		}
		m.id = -1
	}

	return nil
}

func (m *MQBaseModule) Ctx() context.Context {
	return m.ctx
}
func (m *MQBaseModule) genSubTopics() string {
	if m.cfg.HasSubAllTopic {
		return ""
	} else {
		return gstr.Join(m.topics.Slice(), ",")
	}
}

func (m *MQBaseModule) doUnpack(topic string, message []byte) (string, []byte) {
	pattern := `(.*)\[([0-9]+)-([0-9]+)]$`
	if !gregex.IsMatchString(pattern, topic) {
		return topic, message
	}

	matchString, err := gregex.MatchString(pattern, topic)
	if err != nil {
		return "", nil
	}

	topic = matchString[1]
	index := gconv.Int(matchString[2])
	packs := gconv.Int(matchString[3])

	if !m.packs.Contains(topic) || index == 0 {
		m.packs.Set(topic, g.MapStrAny{
			"topic": topic,
			"packs": packs,
			"msgs": garray.NewSortedArray(func(a interface{}, b interface{}) int {
				return a.(g.MapStrAny)["index"].(int) - b.(g.MapStrAny)["index"].(int)
			}),
		})
	}

	pack := m.packs.Get(topic).(g.MapStrAny)
	array := pack["msgs"].(*garray.SortedArray)
	array.Append(g.MapStrAny{
		"index": index,
		"msg":   message,
	})

	if array.Len() >= packs {
		msg := make([]byte, 0, 0)

		for !array.IsEmpty() {
			v, _ := array.PopLeft()
			msg = append(msg, v.(g.MapStrAny)["msg"].([]byte)...)
		}

		m.packs.Remove(topic)
		return topic, msg
	}

	return "", nil
}

// 询问配置中心配置参数
func (m *MQBaseModule) askCfg() {
	if g.IsEmpty(m.AppCfg) && m.cfg.UseConfigCenter {
		m.Pub(m.TopicTool.GenCfgAskTopic(MQTopicInput{}), MQEmptyJsonMessage())
	}
}

func (m *MQBaseModule) heartHandler(topic string, message []byte) {
	models := gjson.New(message)
	if g.IsEmpty(models) {
		m.Log().Warning(m.ctx, "心跳询问话题异常, message 没有消息")
		return
	}

	if garray.NewStrArrayFrom(models.Get(".").Strings(), true).Contains(m.cfg.Name) {
		m.Pub(m.TopicTool.GenHeartTopic(MQTopicInput{}), MQEmptyJsonMessage())
		m.askCfg()
	}
}

func (m *MQBaseModule) logHandler(ctx context.Context, input *glog.HandlerInput) {
	input.Next(ctx)
	_, err := m.cl.Pub(m.TopicTool.GenLogTopic(MQTopicInput{}),
		input.Buffer.Bytes(),
	)
	if err != nil {
		yslog.Error(err)
		return
	}
}

func (m *MQBaseModule) cfgHandler(topic string, msg []byte) {
	el := gstr.SplitAndTrim(topic, ".")
	name := el[len(el)-1]
	if name == cfgAll || name == m.cfg.Name {
		m.Log().Info(m.ctx, "获取到配置数据")
		m.Log().Debug(m.ctx, msg)
		m.AppCfg = gjson.New(msg)
	}
}

func (m *MQBaseModule) okSubHandler(topic string, message []byte) {
	m.id = gjson.New(message).Get("id").Int32()
}

func (m *MQBaseModule) Pub(topic string, message []byte) int32 {

	m.Log().Debugf(m.ctx, "topic: %s, message: %s", topic, string(message))

	pub, err := m.cl.Pub(topic, message)
	if err != nil {
		m.Log().Error(m.ctx, "发布消息异常！", err)
		return 0
	}
	return pub
}

func (m *MQBaseModule) PubJson(topic string, message g.MapStrAny) int32 {
	encode, err := gjson.Encode(message)
	if err != nil {
		m.Log().Error(m.ctx, "转换 JSON 失败！", err)
		return 0
	}

	return m.Pub(topic, encode)
}

func (m *MQBaseModule) PubJsonRes(topic string, code int, message string, data interface{}) int32 {

	return m.PubJson(topic, g.MapStrAny{
		"code":    code,
		"message": message,
		"data":    data,
	})
}

func (m *MQBaseModule) PubJsonResOk(topic string, data interface{}) int32 {

	return m.PubJsonRes(topic, 0, "", data)
}

func (m *MQBaseModule) PubJsonResError(topic string, code int, message string) int32 {

	return m.PubJsonRes(topic, code, message, nil)
}
